
package com.flinkBigWork.util;


import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.util.*;


public class OneSecondRes4 implements AllWindowFunction<Tuple2<Map<String , Integer>, String>, String, TimeWindow> {
    @Override
    public void apply(TimeWindow timeWindow, Iterable<Tuple2<Map<String , Integer>, String>> input, Collector<String> out) {
        List<String> all_codes = Arrays.asList(
                "沙拉窗口",
                "小吃窗口",
                "快餐窗口",
                "烧烤窗口",
                "盖浇饭窗口",
                "饮品窗口",
                "素食窗口",
                "甜品窗口",
                "特色菜窗口",
                "面食窗口"
        );

        if (!input.iterator().hasNext()) {
            return;
        }
        Tuple2<Map<String , Integer>, String> result = input.iterator().next();
        Map<String,  Integer> map1 = result.f0;
        // 调用打印函数
//        printMap(map1, "  "); // 添加前缀以缩进输出
        Map<String, Integer> resString = filterMapByKeys(map1, all_codes);
        List<Integer> list1 = new ArrayList<>(resString.values());
//        System.out.println(list1);

        String time = result.f1;
//        System.out.println(time);
        String listKey = "issue4";


        try (Jedis jedis = jedisPool.getJedis()) {
            if (jedis != null) {
                // 切换到数据库1号
                jedisPool.select_db0(jedis);
                jedis.rpush(listKey, time);
                
//                System.out.println(list1.size());
                for (int i = 0; i < list1.size(); i++) {

                    Integer queueLength = list1.get(i);

                    System.out.println("***********");
                    System.out.println(i);


                    int len = Math.toIntExact(jedis.llen(listKey));
                    // 如果列表长度小于11，则直接添加,这里T=11
                    if (len < 11) {
                        jedis.rpush(listKey, String.valueOf(queueLength));
                    } else {
                        if(queueLength==0){
                            Integer oldQueueLength = Integer.valueOf(jedis.lindex(listKey, -11));
                            queueLength=oldQueueLength;
                            jedis.rpush(listKey, String.valueOf(queueLength));
                        }
                        else{
                            jedis.rpush(listKey, String.valueOf(queueLength));
                        }
                    }
                }
            }else{
                    System.out.println("Jedis is not initialized.");
                }
            } catch(Exception e){
                e.printStackTrace();
            }
        }

        /**
         * 从原始Map中提取特定的键，并创建一个新的Map，只包含这些键及其对应的值。
         * 如果原始Map中不存在某个键，则在新Map中对应的值为0.0。
         *
         * @param map 原始的Map，包含所有键值对。
         * @param keys 需要从原始Map中提取的键的集合。
         * @return 一个新的Map，只包含特定的键及其对应的值。
         */
        private Map<String, Integer> filterMapByKeys(Map<String, Integer> map, List<String> keys) {
            // 使用LinkedHashMap保持键的顺序
            Map<String, Integer> filteredMap = new LinkedHashMap<>();

            // 遍历所有指定的键
            for (String key : keys) {
                // 从原始Map中获取值，如果不存在则使用默认值0.0
                filteredMap.put(key, map.getOrDefault(key, 0));
            }

            return filteredMap;
        }
    public static void printMap(Map<String, Integer> map, String prefix) {
        if (map == null || map.isEmpty()) {
            System.out.println((prefix == null ? "" : prefix) + "Map is empty or null.");
            return;
        }

        System.out.println((prefix == null ? "" : prefix) + "Map contents:");
        for (Map.Entry<String, Integer> entry : map.entrySet()) {
            String key = entry.getKey();
            Integer value = entry.getValue();
            System.out.println((prefix == null ? "" : prefix) + "Key: " + key + ", Value: (" + value + ")");
        }
    }


}
